package org.tao.lightningmq.broker.event.spi.listenr;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.enums.BrokerClusterModeEnum;
import org.tao.lightningmq.broker.event.model.CreateTopicEvent;
import org.tao.lightningmq.broker.model.CommitLogModel;
import org.tao.lightningmq.broker.model.MqTopicModel;
import org.tao.lightningmq.broker.model.QueueModel;
import org.tao.lightningmq.broker.utils.LogFileNameUtil;
import org.tao.lightningmq.common.constants.BrokerConstants;
import org.tao.lightningmq.common.dto.CreateTopicReqDTO;
import org.tao.lightningmq.common.dto.CreateTopicRespDTO;
import org.tao.lightningmq.common.enums.BrokerEventCode;
import org.tao.lightningmq.common.enums.BrokerResponseCode;
import org.tao.lightningmq.common.event.Listener;
import org.tao.lightningmq.common.tcp.TcpMsg;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author lht
 * @date 2025/1/21 - 12:50
 * @description:
 */
public class CreateTopicListener implements Listener<CreateTopicEvent> {

    private static final Logger LOGGER = LoggerFactory.getLogger(CreateTopicListener.class);

    @Override
    public void onReceive(CreateTopicEvent event) throws Exception {
        CreateTopicReqDTO createTopicReqDTO = event.getCreateTopicReqDTO();
        if (createTopicReqDTO.getQueueSize() < 0 || createTopicReqDTO.getQueueSize() > 20) {
            throw new IllegalArgumentException("queueSize参数异常");
        }
        MqTopicModel mqTopicModel = CommonCache.getMqTopicModelMap().get(createTopicReqDTO.getTopic());
        if (mqTopicModel != null) {
            throw new IllegalArgumentException("topic已经存在");
        }
        createTopicFile(createTopicReqDTO);
        addTopicInCommonCache(createTopicReqDTO);
        loadFileInMMap(createTopicReqDTO);
        LOGGER.info("topic:{} is created! queueSize is {}", createTopicReqDTO.getTopic(), createTopicReqDTO.getQueueSize());
        CreateTopicRespDTO createTopicRespDTO = new CreateTopicRespDTO();
        createTopicRespDTO.setMsgId(event.getMsgId());
        event.getContext().write(new TcpMsg(BrokerResponseCode.CREATED_TOPIC_SUCCESS.getCode(), JSON.toJSONBytes(createTopicRespDTO)));
        if(BrokerClusterModeEnum.MASTER_SLAVE.getCode().equals(CommonCache.getGlobalProperties().getBrokerClusterMode())
                && "master".equals(CommonCache.getGlobalProperties().getBrokerClusterRole())) {
            LOGGER.info("主节点将创建topic请求同步给到从节点");
            //主节点，需要将创建topic请求同步给到从节点
            for (ChannelHandlerContext slaveChannel : CommonCache.getSlaveChannelMap().values()) {
                slaveChannel.writeAndFlush(new TcpMsg(BrokerEventCode.CREATE_TOPIC.getCode(), JSON.toJSONBytes(createTopicReqDTO)));
            }
        }
    }

    private void loadFileInMMap(CreateTopicReqDTO createTopicReqDTO) throws Exception {
        CommonCache.getCommitLogAppendHandler().prepareMMapLoading(createTopicReqDTO.getTopic());
        CommonCache.getConsumeQueueAppendHandler().prepareConsumeQueue(createTopicReqDTO.getTopic());
    }

    private void addTopicInCommonCache(CreateTopicReqDTO createTopicReqDTO) {
        MqTopicModel mqTopicModel = new MqTopicModel();
        mqTopicModel.setTopic(createTopicReqDTO.getTopic());
        long currentTimeMillis = System.currentTimeMillis();
        mqTopicModel.setCreateAt(currentTimeMillis);
        mqTopicModel.setUpdateAt(currentTimeMillis);
        CommitLogModel commitLogModel = new CommitLogModel();
        commitLogModel.setFileName(LogFileNameUtil.getFirstCommitLogName());
        commitLogModel.setOffset(new AtomicInteger(0));
        commitLogModel.setOffsetLimit(BrokerConstants.COMMIT_LOG_DEFAULT_MMAP_SIZE.longValue());
        mqTopicModel.setCommitLogModel(commitLogModel);
        List<QueueModel> queueModelList = new ArrayList<>(createTopicReqDTO.getQueueSize());
        mqTopicModel.setQueueList(queueModelList);
        for (int i = 0; i < createTopicReqDTO.getQueueSize(); i++) {
            QueueModel queueModel = new QueueModel();
            queueModel.setId(i);
            queueModel.setFileName(LogFileNameUtil.getFirstConsumeQueueName());
            queueModel.setOffsetLimit(BrokerConstants.CONSUME_QUEUE_DEFAULT_MMAP_SIZE);
            queueModel.setLastOffset(0);
            queueModel.setLatestOffset(new AtomicInteger(0));
            queueModelList.add(queueModel);
        }
        CommonCache.getTopicModelList().add(mqTopicModel);
    }

    private void createTopicFile(CreateTopicReqDTO createTopicReqDTO) throws IOException {
        String commitLogDirPath = LogFileNameUtil.buildCommitLogBasePath(createTopicReqDTO.getTopic());
        new File(commitLogDirPath).mkdirs();
        new File(commitLogDirPath + BrokerConstants.SPLIT +
                LogFileNameUtil.getFirstCommitLogName()).createNewFile();
        String consumeQueueBasePath = LogFileNameUtil.buildConsumeQueueBasePath(createTopicReqDTO.getTopic());
        new File(consumeQueueBasePath).mkdirs();
        for (int i = 0; i < createTopicReqDTO.getQueueSize(); i++) {
            String path = consumeQueueBasePath + BrokerConstants.SPLIT + i;
            new File(path).mkdirs();
            new File(path + BrokerConstants.SPLIT +
                    LogFileNameUtil.getFirstConsumeQueueName()).createNewFile();
        }
    }
}
